package za.org.jcicada.comms.impl.katcp;

/**
 * KatcpCommsBuffer.java
 *
 * @author Simon Cross
 *
 * Created on 20 February 2009, 11:11
 */

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.CharsetEncoder;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import za.org.jcicada.comms.ICommsBuffer;
import za.org.jcicada.comms.IKatcpSerializerStrategy;
import za.org.jcicada.comms.messages.CommsMessage;
import za.org.jcicada.katcp.Message;
import za.org.jcicada.katcp.Parser;

/**
 * A buffering mechanism for a KatcpConnection.
 *
 * A KatcpCommsBuffer is fed data from a stream of bytes (e.g. a tcp connection)
 * and adds completed CommsMessages received to a message queue. In addition,
 * it provides a utility method for converting a CommsMessage to a stream of
 * bytes for sending.
 *
 * Messages are encoded by generating a KATCP message using the IKatcpSerializer
 * provided and then converting the KATCP message to bytes using the standard
 * KATCP wire representation.
 *
 * The toByteBuffer method is used to encode messages to a byte buffer for
 * sending. feedByteBuffer processes a chunk of data. Completed messages can be
 * retrieved using popMessage which returns null if there are no completed
 * messages available.
 */
public class KatcpCommsBuffer implements ICommsBuffer {

    private static Log logger = LogFactory.getLog(KatcpCommsBuffer.class);

    /**
     * Serializer for converting messages to bytes and back again.
     */
    protected IKatcpSerializerStrategy serializer;

    /**
     * Decoder for decoding incoming message strings.
     */
    protected static CharsetDecoder wireDecoder = Charset.forName("ascii").newDecoder();

    /**
     * Encoder for encoding outgoing message strings.
     */
    protected static CharsetEncoder wireEncoder = Charset.forName("ascii").newEncoder();

    /**
     * Parser instance for parsing strings into Message objects.
     */
    protected final Parser parser = new Parser();

    /**
     * Queue for messages.
     */
    protected ConcurrentLinkedQueue<CommsMessage> messageQueue;

    /**
     * Unprocessed data (String).
     */
    protected String chunk;

    /**
     * Create a KatcpCommsBuffer object.
     */
    public KatcpCommsBuffer(IKatcpSerializerStrategy serializer) {
        this.serializer = serializer;
        messageQueue = new ConcurrentLinkedQueue<CommsMessage>();
        chunk = "";
    }

    /**
     * Convert a message to a byte buffer containing a KATCP message string.
     */
    @Override
    public ByteBuffer toByteBuffer(CommsMessage message) throws IOException {
        Message katcpMsg = serializer.toKatcp(message);
        ByteBuffer data;
        data = wireEncoder.encode(CharBuffer.wrap(katcpMsg.toString() + "\n"));
        data.position(data.limit()); // set data to it's limit so it's ready to flip
        return data;
    }

    /**
     * Handle a chunk of data. Ported from the katcp BaseServer handleChunk method.
     *
     * @param chunk A piece of data from the client to process.
     * @return The unprocessed part of the data.
     * @throws IOException If the attempt to process the chunk fails.
     */
    public String handleChunk(String chunk) throws IOException {
        chunk = chunk.replace('\r', '\n');
        String [] lines = chunk.split("\\n", -1);
        String fullLine;
        Message katcpMsg;
        CommsMessage commsMsg;

        for (int i=0; i < lines.length-1; i++) {
            fullLine = lines[i];
            if (fullLine.length() == 0)
                continue;
            try {
                katcpMsg = parser.parse(fullLine);
                commsMsg = serializer.fromKatcp(katcpMsg);
                messageQueue.add(commsMsg);
            }
            catch (Exception ex) {
                logger.error("Failed to parse message", ex);
            }
        }

        return lines[lines.length-1];        
    }

    /**
     * Process a ByteBuffer, adding any messages to the internal message queue.
     */
    @Override
    public void feedByteBuffer(ByteBuffer buffer) throws IOException {
        CharBuffer cbuffer;
        try {
            cbuffer = wireDecoder.decode(buffer);
        } catch (CharacterCodingException ex) {
            throw new IOException("Failed to decode KATCP data to string.", ex);
        }
        chunk = handleChunk(chunk + cbuffer.toString());
    }

    /**
     * Pop a message from the internal message queue. Return null if the queue 
     * is empty.
     */
    @Override
    public CommsMessage popMessage() {
        return messageQueue.poll();
    }

}
